package cn.doitedu.rtmk.demo7;

import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;

import java.io.IOException;

public class RuleModel_C_Calculator implements RuleCalculator {

    private static final Logger log = LoggerFactory.getLogger("cn.doitedu.rtmk.demo7.RuleModel_C_Calculator");

    ValueState<Integer> cntState;
    Jedis jedis;
    String ruleId;
    JSONObject paramJsonObject;
    JSONObject message;
    Long histQueryEndTime;  // 预计算时的分界点

    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJson, Roaring64Bitmap preSelectUsersBitmap, Long histQueryEndTime) throws Exception {
        paramJsonObject = JSON.parseObject(ruleParamJson);
        ruleId = paramJsonObject.getString("rule_id");
        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId + "-cnt-state", Integer.class));

        this.histQueryEndTime = histQueryEndTime;

        jedis = new Jedis("doitedu", 6379);

        message = new JSONObject();
        message.put("rule_id", ruleId);

    }

    @Override
    public void calculate(UserEvent userEvent, Collector<String> out) throws IOException {
        // 判断进来的数据，是否是历史分界点之前的数据；如果是，则直接返回；
        log.warn("收到数据，数据时间:{},历史分界点时间:{}",userEvent.getAction_time(),histQueryEndTime);
        if (userEvent.getAction_time() <= histQueryEndTime) return;

        long userId = userEvent.getUser_id();
        String eventId = userEvent.getEvent_id();
        JSONObject dynamicProfileCondition = paramJsonObject.getJSONObject("dynamic_profile_condition");
        String conditionEventId = dynamicProfileCondition.getString("event_id");
        int compareValue = dynamicProfileCondition.getIntValue("compare_value");
        long startTime = dynamicProfileCondition.getLongValue("start_time");
        long endTime = dynamicProfileCondition.getLongValue("end_time");


        /* *
         * 一、动态画像统计
         */
        if (eventId.equals(conditionEventId) && userEvent.getAction_time() < (endTime== -1? Long.MAX_VALUE : endTime )  ) {
            /* *
             * 取oldValue
             */
            int oldValue = 0;

            // 1. 先从状态中取oldValue
            Integer stateOldValue = cntState.value();
            if (stateOldValue == null) {
                // 2. 从redis中去获取
                // redis中的历史值结构：
                //         大key:   rule_id
                //         小key:   user_id  => value: 历史值
                String redisOldValueStr = jedis.hget(ruleId, userId + "");
                log.warn("查询redis中的历史统计值,查询结果:{}",redisOldValueStr);

                if (redisOldValueStr != null) {
                    oldValue = Integer.parseInt(redisOldValueStr);
                }
            } else {
                oldValue = stateOldValue;
            }

            // 在oldValue基础上+1，并放入状态
            cntState.update(oldValue + 1);

        }

        /* *
         * 二、触发判断
         */
        String triggerEventId = paramJsonObject.getString("trigger_event_id");
        if (eventId.equals(triggerEventId)) {

            // 获取动态画像统计条件的统计结果
            Integer curValue = cntState.value();
            if(curValue == null ){
                // 去redis中获取历史值
                String redisStr = jedis.hget(ruleId, userId + "");
                if(redisStr == null ){
                    curValue = 0;
                }else{
                    curValue = Integer.parseInt(redisStr);
                }
            }


            if (curValue >= compareValue) {
                message.put("user_id", userId);
                message.put("hit_time", userEvent.getAction_time());

                out.collect(message.toJSONString());
            }
        }


    }
}
